package com.thunderbit.kafka.consumer;

import com.thunderbit.kafka.config.KafkaConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * Kafka 简单消费者示例
 * 演示如何从 Kafka 消费消息
 */
public class SimpleConsumer {
    
    private static final Logger logger = LoggerFactory.getLogger(SimpleConsumer.class);
    
    public static void main(String[] args) {
        // 1. 获取配置
        Properties props = KafkaConfig.getConsumerProperties();
        
        // 2. 创建 Consumer 实例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        
        try {
            // 3. 订阅主题
            consumer.subscribe(Collections.singletonList(KafkaConfig.TOPIC_NAME));
            logger.info("已订阅主题: {}", KafkaConfig.TOPIC_NAME);
            
            // 4. 持续拉取消息
            while (true) {
                // 拉取消息，超时时间 100 毫秒
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                // 处理每条消息
                for (ConsumerRecord<String, String> record : records) {
                    logger.info("收到消息 - Topic: {}, Partition: {}, Offset: {}, Key: {}, Value: {}", 
                                record.topic(),
                                record.partition(),
                                record.offset(),
                                record.key(),
                                record.value());
                }
                
                // 可选：手动提交偏移量（如果禁用了自动提交）
                // consumer.commitSync();
            }
            
        } catch (Exception e) {
            logger.error("消费消息时发生错误", e);
        } finally {
            // 5. 关闭 Consumer
            consumer.close();
            logger.info("Consumer 已关闭");
        }
    }
}